package event;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import parallel.BasicRangeInputSplit;
import org.json.simple.JSONValue;
public class PageGenerator extends Generator{
	private MultipleOutputs<Text,NullWritable> output = null;
	private String type = null;
	private BasicRangeInputSplit inputSplit = null;
	private static final String DATABASE =  "Database";
	private static final String ID = "Id";
	private static final String LINK = "Link";
	public PageGenerator(MultipleOutputs<Text,NullWritable> output, BasicRangeInputSplit inputSplit,int numOfRegions, String type)
	{
		super(numOfRegions, inputSplit);
		this.inputSplit = inputSplit;
		this.output = output;
		this.type = type;
	}
	public void generate(long inputKey, String inputValue) throws IOException, InterruptedException {
		Basic basic = basicConfig.getBasic(type);
		EventNode node = nodeConfig.getEventNode(type);
		int format = basicConfig.getFormat();
		if(format == 1)
		{
			generateCsv(basic, node, inputKey);
		}
		else
		{
			generateJson(basic, node, inputKey);
		}
	}
	private void generateCsv(Basic basic, EventNode node, long inputKey) throws IOException, InterruptedException
	{
		List<Attribute> attrList = basic.getAttrList();
		String name = basic.getName();
		StringBuffer sb = new StringBuffer();
		int splitId = inputSplit.getSplitId();
		int numPerRegion = (int)(basic.getNumber()/numOfRegions);
		long currId = splitId * numPerRegion + inputKey;
		DecimalFormat formatter = new java.text.DecimalFormat("000000000000");
		String id = formatter.format(currId);
		sb.append(name + "-"+ id);
		//store the attribute names and values already generated for this page
		Map<String, String> attrValues = new HashMap<String, String>();
		attrValues.put(ID, Long.toString(currId));
		for(int i = 0; i < attrList.size(); i++)
		{
			Attribute attr = attrList.get(i);
			String value = getAttrValue(basic,attr,attrValues);
			attrValues.put(attr.getName(), value);
			sb.append("," + value);
		}
		// write the basic attributes of a page in one file
		String outputPath = name + "/" + name;
		output.write(new Text(sb.toString()), NullWritable.get(), outputPath);
		List<Link> linkList = node.getLinkList();
		Random random = new Random();
		for(int i = 0; i < linkList.size(); i++)
		{
			Link link = linkList.get(i);
			String linkType = link.getLinkType();
			int min = link.getMin();
			int max = link.getMax();
			// how many links this page has
			int linkNum = min + random.nextInt(max - min + 1);
			Set<String> linkSet = new HashSet<String>();
			// generate the each link
			for(int k = 0; k < linkNum; k++)
			{
				StringBuffer linkBuffer = new StringBuffer();
				linkBuffer.append(name + "-" + id);
				AttrGenerator attrGen = attrGenerators.get(linkType);
				List<String> params = link.getParams(); 
				String linkValue = attrGen.generate(params);
				// if we have already generate a link with the same value, we skip it
				while(linkSet.contains(linkValue))
				{
					linkValue = attrGen.generate(params);
				}
				linkBuffer.append("," + linkValue);
				List<Attribute> linkAttrList = link.getAttrs();
				//generate attributes belongs to this link
				for(int j = 0; j < linkAttrList.size(); j++)
				{
					Attribute linkAttr = linkAttrList.get(j);
					String linkAttrValue = getAttrValue(basic, linkAttr, attrValues);
					linkBuffer.append("," + linkAttrValue);
				}
				String linkName = link.getLinkName();
				String linkOutputPath = LINK +"/" + name + linkName + "/" + name + "-" + linkName;
				// write each type of link into a separate folder
				output.write(new Text(linkBuffer.toString()), NullWritable.get(), linkOutputPath);
			}
		}
	}
	// generate attributes based on the attribute type with an attribute generator
	private String getAttrValue(Basic basic, Attribute attr, Map<String, String> attrValues)
	{
		Map<String, Attribute> attrMap = basic.getAttrMap();
		String attrType = attr.getType();
		List<String> params = attr.getParams();
		List<String> inputParams = new LinkedList<String>();
		inputParams.addAll(params);
		AttrGenerator attrGen = attrGenerators.get(attrType);
		//if the attribute depends on other attribute already generated 
		// and it is from a database table 
		if(attrType.equals(DATABASE) && inputParams.size() > 1)
		{
			String relatedField = inputParams.get(1);
			String fieldValue = attrValues.get(relatedField);
			// we use the value of the field as a parameter. 
			// Here we assume (related attribute value/maximum related value) 
			// in the same range as (current attribute value / maximum current attribute value)
			inputParams.set(1, fieldValue);
			if(relatedField.equals(ID))
			{
				long number = basic.getNumber();
				inputParams.add(Long.toString(number));
			}
			else
			{
				String total = attrMap.get(relatedField).getParams().get(0);
				inputParams.add(total);
			}		
		}
		String value = attrGen.generate(inputParams);
		return value;
	}
	// The implementation of method is very similar to the CSV version
	private void generateJson(Basic basic, EventNode eventNode, long inputKey) throws IOException, InterruptedException {
		Map nodeJson = new HashMap();
		List<Attribute> attrList = basic.getAttrList();
		String name = basic.getName();
		int splitId = inputSplit.getSplitId();
		int numPerRegion = (int)(basic.getNumber()/numOfRegions);
		long currId = splitId * numPerRegion + inputKey;
		DecimalFormat formatter = new java.text.DecimalFormat("000000000000");
		String id = formatter.format(currId);
		nodeJson.put(ID, name + "-" + id);
		Map<String, String> attrValues = new HashMap<String, String>();
		attrValues.put(ID, Long.toString(currId));
		for(int i = 0; i < attrList.size(); i++)
		{
			Attribute attr = attrList.get(i);
			String attrValue = getAttrValue(basic,attr, attrValues);
			attrValues.put(attr.getName(), attrValue);
			String attrName = attr.getName();
			nodeJson.put(attrName, attrValue);
		}
		List<Link> linkList = eventNode.getLinkList();
		Random random = new Random();
		for(int i = 0; i < linkList.size(); i++)
		{
			Link link = linkList.get(i);
			List linkValueList = new ArrayList();
			String linkName = link.getLinkName();
			nodeJson.put(linkName, linkValueList);
			String linkType = link.getLinkType();
			int min = link.getMin();
			int max = link.getMax();
			int linkNum = min + random.nextInt(max - min + 1);
			Set<String> linkSet = new HashSet<String>();
			for(int k = 0; k < linkNum; k++)
			{
				Map linkNameValue = new HashMap();
				AttrGenerator attrGen = attrGenerators.get(linkType);
				List<String> params = link.getParams(); 
				String linkValue = attrGen.generate(params);
				while(linkSet.contains(linkValue))
				{
					linkValue = attrGen.generate(params);
				}
				linkNameValue.put(params.get(0), linkValue);
				List<Attribute> linkAttrList = link.getAttrs();
				for(int j = 0; j < linkAttrList.size(); j++)
				{
					Attribute linkAttr = linkAttrList.get(j);
					String linkAttrValue = getAttrValue(basic, linkAttr, attrValues);
					linkNameValue.put(linkAttr.getName(), linkAttrValue);
				}
				linkValueList.add(linkNameValue);
			}
		}		
		String text = JSONValue.toJSONString(nodeJson);
		String outputPath = LINK + "/" + name + "/" + name;
		output.write(new Text(text), NullWritable.get(), outputPath);
	}
}
